package cn.ljh.demo;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerOrderly;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.CommandLineRunner;
import org.springframework.stereotype.Component;

import java.nio.charset.StandardCharsets;
import java.util.List;

/**
 * 自定义监听器 -- 鉴定crud后传来的假订单消息
 * <p>
 * CommandLineRunner :当 Spring Boot 项目启动完成（即所有 Bean 加载完成后），就会执行 run() 方法中的逻辑。
 *
 * @author lujinhong
 * @since 2025-05-30
 */
@Component
public class Consumer03 implements CommandLineRunner {

    // 拿配置文件的地址
    @Value("${rocketmq.name-server}")
    private String nameServer;

    @Override
    public void run(String... args) throws Exception {
        // 创建消费者实例
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("ljh_consumer_group");
        // 设置 nameserver 的地址
        consumer.setNamesrvAddr(nameServer);
        // 设置消息消费失败后的最大重试次数，超过3次则将消息存到死信队列中（默认是16次）
        consumer.setMaxReconsumeTimes(3);
        // 订阅消息的主题和标签
        consumer.subscribe("LjhTopic01", "Tag02");

        // 注册监听器 -- MessageListenerOrderly 顺序消费
        consumer.registerMessageListener(new MessageListenerOrderly() {
            @Override
            public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgList, ConsumeOrderlyContext consumeOrderlyContext) {
                // 遍历 broker 发来的消息
                for (MessageExt msg : msgList) {
                    String str = new String(msg.getBody(), StandardCharsets.UTF_8);
                    System.err.println("获取消息体：" + str);
                }

                // 告诉 RocketMQ 消费者 这批消息已经成功消费，可以正常提交消费进度了
                return ConsumeOrderlyStatus.SUCCESS;
            }
        });
        // 启动消费者
        consumer.start();
        System.err.println("启动消费者-----------------------");
    }
}
